package org.tomac.io.fix;

import java.io.IOException;
import java.net.MalformedURLException;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.log4j.Logger;
import org.tomac.io.IConnection;
import org.tomac.io.IIOPlexor;
import org.tomac.io.ILogHandle;
import org.tomac.io.ITimerInstance;
import org.tomac.io.ITimerTask;
import org.tomac.protocol.fix.FixGarbledException;
import org.tomac.protocol.fix.FixSessionException;
import org.tomac.protocol.fix.FixUtils;
import org.tomac.protocol.fix.messaging.FixHeartbeat;
import org.tomac.protocol.fix.messaging.FixLogon;
import org.tomac.protocol.fix.messaging.FixLogout;
import org.tomac.protocol.fix.messaging.FixMessage;
import org.tomac.protocol.fix.messaging.FixMessageInfo;
import org.tomac.protocol.fix.messaging.FixMessageInfo.MsgTypes;
import org.tomac.protocol.fix.messaging.FixMessageInfo.SessionRejectReason;
import org.tomac.protocol.fix.messaging.FixMessageListener;
import org.tomac.protocol.fix.messaging.FixMessageListenerImpl;
import org.tomac.protocol.fix.messaging.FixMessageParser;
import org.tomac.protocol.fix.messaging.FixReject;
import org.tomac.protocol.fix.messaging.FixResendRequest;
import org.tomac.protocol.fix.messaging.FixSequenceReset;
import org.tomac.protocol.fix.messaging.FixSessionValidator;
import org.tomac.protocol.fix.messaging.FixTags;
import org.tomac.protocol.fix.messaging.FixTestRequest;
import org.tomac.utils.Utils;

public abstract class FixSessionBase extends FixMessageListenerImpl implements IFixSession {
	
	private static Logger log = Logger.getLogger(FixSessionBase.class.getSimpleName());
	
	protected int inMsgSeqNum = 0;
	protected int outMsgSeqNum = 0;
	private int resendRequestToo = 0;
	
	private ByteBuffer buf = ByteBuffer.allocate(1024);
	private ByteBuffer resendBuf = ByteBuffer.allocate(1024);
	
	protected IFixApplication app;
	private FixSessionBase in;
	protected OutboundFixMessageListener out;
	
	private FixMessageParser parser;
	private FixMessageParser resendParser;
	
	private final int SEQUENCERESET_INT = 52;
	
	private FixSessionState fixSessionState = FixSessionState.DISCONNECTED_NO_CONNECTION_TODAY;
	protected IIOPlexor plex;
	protected IFixURL url;
	protected IConnection connection;
	protected ILogHandle outboundLogHandle;
	protected ILogHandle inboundLogHandle;
	private FixMessageListenerImpl resendListener;
	
	private FixReject reject = new FixReject();
	private FixLogon logon = new FixLogon();
	private FixLogout logout = new FixLogout();
	private FixHeartbeat heartbeat = new FixHeartbeat();
	private FixTestRequest testrequest = new FixTestRequest();
	private FixResendRequest resendrequest = new FixResendRequest();
	private FixSequenceReset sequencereset = new FixSequenceReset();
	
	public HbtTimerTask hbtTimerTask;
	public ITimerInstance hbtTimerInstance;
	public TestRequestTimerTask testRequestTimerTask;
	public ITimerInstance testRequestTimerInstance;
	private DisconnectTimerTask logoutTimerTask;
	private ITimerInstance logoutTimerInstance;
	protected DisconnectTimerTask disconnectNoLogonTimerTask;
	protected ITimerInstance disconnectNoLogonTimerInstance;
	
	Queue<ByteBuffer> pendingInboundMsg;
	
	private long lastInMsgTimeInMillis = 0;
	private long lastOutMsgTimeInMillis = 0;
	private int outstandingHeartbeats = 0;
	
	public FixSessionBase(IIOPlexor plex, IFixURL url, IFixApplication application) {
		this.plex = plex;
		this.url = url;
		app = application;
		in = this;
		testRequestTimerTask = new TestRequestTimerTask();
		hbtTimerTask = new HbtTimerTask();
		logoutTimerTask = new DisconnectTimerTask();
		disconnectNoLogonTimerTask = new DisconnectTimerTask();
		parser = new FixMessageParser();
		parser.validator = new FixSessionValidatorImp();
		resendParser = new FixMessageParser();
		pendingInboundMsg = new LinkedList<ByteBuffer>();
		resendListener = new FixMessageListenerImpl();
	}
	
	protected void setFixSessionState(FixSessionState state) {
		log.info(fixSessionState + "->" + state);
		fixSessionState = state;
	}
	
	public void incrementInMsgSeqNum() {
		inMsgSeqNum++;
		lastInMsgTimeInMillis = plex.currentTimeMillis();
		if (log.isDebugEnabled()) log.debug("inMsgSeqNum: " + inMsgSeqNum);
	}

	void incrementOutMsgSeqNum() {
		lastOutMsgTimeInMillis = plex.currentTimeMillis();
		outMsgSeqNum++;
		if (log.isDebugEnabled()) log.debug("outMsgSeqNum: " + outMsgSeqNum);
	}
	
	@Override
	public void onDisconnect() {
		setFixSessionState(FixSessionState.DETECT_BROKEN_NETWORK_CONNECTION);
		app.onDisconnect();
		setFixSessionState(FixSessionState.DISCONNECTED_CONNECTION_TODAY);
		if (testRequestTimerInstance!=null)  testRequestTimerInstance.cancel(testRequestTimerTask); 
		if (hbtTimerInstance!=null) hbtTimerInstance.cancel(hbtTimerTask); 
	}

	@Override
	public void onRead(ByteBuffer buf) {

		if (log.isDebugEnabled()) log.debug("read message: " + (buf.limit() - buf.position()) + " bytes");
        
		if (Boolean.getBoolean("fix.loginbound")) {
			try {
				inboundLogHandle.write((ByteBuffer)buf.slice());
			} catch (IOException e) {
				if (log.isDebugEnabled()) log.error(e.getMessage(), e);
				else log.error(e.getMessage());
			}
		}

		while (buf.hasRemaining()) {
			int read = _onRead(buf.slice());
			if (read == 0) return;
			buf.position(buf.position() + read);
		}
		
	}
	
	private int _onRead(ByteBuffer buf) {

		if (fixSessionState == FixSessionState.NETWORK_CONNECTION_ESTABLISHED)
			disconnectNoLogonTimerInstance.cancel(disconnectNoLogonTimerTask);
		
		try {
			FixMessage msg = parser.parse(buf, app.getInboundFixMessageReplicator());
			if (msg == null) return 0; // indicate buffer does not contain full fix message.
		} catch (FixSessionException e) {

			log.error(e.getMessage());

			buf.position(buf.limit()); //pretend we read all
			if (getFixSessionState() == FixSessionState.ACTIVE || getFixSessionState() == FixSessionState.PROCESSING_RESPONSE_TO_RESENDREQUEST) {
				log.error("send reject");
				app.getInboundFixMessageReplicator().onFixReject(createMagicReject(e));
			} else {
				log.error("close connection");
				connection.close();
			}
		} catch (FixGarbledException e) {
			if (fixSessionState == FixSessionState.NETWORK_CONNECTION_ESTABLISHED)
				connection.close();
			// this message is garbled ignore it, but pretend we read all
			buf.position(buf.limit());
			if (log.isDebugEnabled()) log.debug(e.getMessage(), e);
			else log.info(e.getMessage());
		}

		if (FixSessionState.PROCESSING_RESPONSE_TO_RESENDREQUEST == getFixSessionState() && inMsgSeqNum >= resendRequestToo) {
			setFixSessionState(FixSessionState.ACTIVE);
			while (!pendingInboundMsg.isEmpty() && FixSessionState.PROCESSING_RESPONSE_TO_RESENDREQUEST != getFixSessionState()) {
				resendRequestToo = -1;
				onRead(pendingInboundMsg.poll());
			}
		}
		return buf.limit();
	}
	
	@Override
	public void onWrite() {};

	@Override
	public FixSessionState getFixSessionState() {
		return fixSessionState;
	}
	

	@Override
	abstract public void onConnect();
	
	@Override
	abstract public void connect() throws IOException;
	
	@Override
	public FixMessageListener getInboundFixMessageListener() {
		return in;
	}

	@Override
	public void send(FixMessage msg) {
		buf.clear();
		populateHeaderForSend(msg);
		msg.encode(buf);
		try {
			outboundLogHandle.write(buf.slice());
			connection.write(buf);
			incrementOutMsgSeqNum();
		} catch (IOException e) {
			if (log.isDebugEnabled()) log.error(e.getMessage(), e);
			else log.error(e.getMessage());
		}
	}

	// PRIVATE METHODS

	private boolean isReaplayableMessage(int msgType) {
		switch(msgType) {
		case MsgTypes.HEARTBEAT_INT:
		case MsgTypes.LOGON_INT:
		case MsgTypes.LOGOUT_INT:
		case MsgTypes.RESENDREQUEST_INT:
		case MsgTypes.TESTREQUEST_INT:
		case MsgTypes.SEQUENCERESET_INT:
			return false;
		}
		return true;
	}

	protected void resend(FixMessage msg) throws IOException {
		buf.clear();
		populateHeaderForResend(msg);
		msg.encode(buf);
		connection.write(buf);
	}
	
	private void populateHeaderForSend(FixMessage msg) {
		Utils.copy(msg.targetCompID, url.getSenderCompID());
		Utils.copy(msg.senderCompID, url.getTargetCompID());
		msg.msgSeqNum = outMsgSeqNum + 1;
		Utils.utcTimestampConverter.convertToUtcTimestamp(msg.sendingTime, plex.currentTimeMillis(), true);
	}

	private void populateHeaderForResend(FixMessage msg) {
		Utils.copy(msg.targetCompID, url.getSenderCompID());
		Utils.copy(msg.senderCompID, url.getTargetCompID());
		Utils.copy(msg.origSendingTime, msg.sendingTime); 
		msg.possDupFlag = true;
		Utils.utcTimestampConverter.convertToUtcTimestamp(msg.sendingTime, plex.currentTimeMillis(), true);
	}
	
	// CREATE HELPER METHODS

	private FixReject createMagicReject(FixSessionException e) {
		// Last byte in text is SOH, is magic to tell that this Reject is internally generated.
		FixReject r = createReject(e.sessionRejectReason, e.text, e.refTagID, e.refMsgType);
		r.text[r.text.length - 1] = FixUtils.SOH;
		return r;
	}
	
	public FixReject createReject(long sessionRejectReason, byte[] text, long tag, byte[] msgType) {
		reject.clear();
		if (FixUtils.isSet(sessionRejectReason)) reject.sessionRejectReason = sessionRejectReason;
		if (FixUtils.isSet(text)) Utils.copy(reject.text, text);
		if (FixUtils.isSet(tag)) reject.refTagID = tag;
		if (msgType.length > 0 && FixUtils.isSet(msgType)) Utils.copyTrim(reject.refMsgType, msgType);
		reject.refSeqNum = inMsgSeqNum + 1;
		return reject;
	}
	
	public FixHeartbeat createHeartbeat(byte[] testReqID) {
		Utils.copy(heartbeat.testReqID, testReqID);
		return heartbeat;
	}

	public FixTestRequest createTestRequest(byte[] testReqID) {
		Utils.copy(testrequest.testReqID, testReqID);
		return testrequest;
	}
	
	public FixResendRequest createResendRequest(long beginSeqNo) {
		resendrequest.beginSeqNo = beginSeqNo;
		resendrequest.endSeqNo = 0;
		return resendrequest;
	}

	public FixSequenceReset createGapFill(long newSeqNo, long msgSeqNum) {
		sequencereset.msgSeqNum = msgSeqNum;
		sequencereset.newSeqNo = newSeqNo + 1;
		sequencereset.gapFillFlag = true;
		return sequencereset;
	}
	
	protected FixLogon createLogon() {
		logon.encryptMethod = FixMessageInfo.EncryptMethod.NONE__OTHER;
		logon.heartBtInt = url.getHeartBtInt() / 1000;
		return logon;
	}

	protected FixLogout createLogout(byte[] text) {
		logout.clear();
		Utils.copy(logout.text, text);
		return logout;
	}
	
	private boolean isMagic(byte[] text) {
		if (text[text.length - 1] == FixUtils.SOH) {
			text[text.length -1] = (byte)0;
			return true;
		}
		return false;
	}
	
	private class FixSessionValidatorImp extends FixSessionValidator {
		
		@Override
		public boolean validate(FixMessage msg) throws FixSessionException {
			if (msg.msgSeqNum > inMsgSeqNum + 1 && msg.msgType != SEQUENCERESET_INT) {
				if (msg.msgType == MsgTypes.LOGON_INT) {
					setFixSessionState(FixSessionState.INITIATION_LOGON_RECEIVED);
					out.onFixLogon(createLogon());
				} 
				if (getFixSessionState() == FixSessionState.PROCESSING_RESPONSE_TO_RESENDREQUEST) {
					// save it for later processing
					pendingInbound(msg.buf);
				} else {
					setFixSessionState(FixSessionState.RECEIVE_MSGSEQNUM_TOO_HIGH);
					pendingInbound(msg.buf);
					out.onFixResendRequest(createResendRequest(inMsgSeqNum + 1));
					resendRequestToo = (int) msg.msgSeqNum - 1;
				}
				return false;
			} else if (msg.msgSeqNum < inMsgSeqNum + 1 && !msg.possDupFlag && msg.msgType != SEQUENCERESET_INT) {
				if (msg.msgType == MsgTypes.LOGON_INT) {
					// handle this in onLogon, need to recheck msgSeqNum
				} else {
					setFixSessionState(FixSessionState.INITIATE_LOGOUT_PROCESS); 
					// THIS IS UNCILLOCITED. QUESTIONABLE IF IT SHOULD BE SUPPORTED
					out.onFixLogout(createLogout(("MsgSeqNum too low, expecting " + (inMsgSeqNum+1) + " but received " + msg.msgSeqNum).getBytes()));
					return false;
				}
			} else if (msg.msgSeqNum < inMsgSeqNum + 1 && msg.possDupFlag && msg.msgType != SEQUENCERESET_INT) { 
				// fix spec ambvivalent, ignor messages regardless
				return false;
			} else if (msg.possDupFlag) {
				if (msg.possDupFlag && Boolean.getBoolean("fix.doSendingTimeCheck")) {
					if (!FixUtils.isSet(msg.origSendingTime)) {
						// this rej shall consume inSeqNo -> in.onMessage
						throw new FixSessionException(SessionRejectReason.REQUIRED_TAG_MISSING, "Required tag missing".getBytes(), FixTags.ORIGSENDINGTIME_INT, FixUtils.getMsgType(msg.msgType));
					} else {
						long origSendingTime = Utils.utcTimestampConverter.convert(msg.origSendingTime).getTime();
						long sendingTime = Utils.utcTimestampConverter.convert(msg.sendingTime).getTime();
						if ( origSendingTime > sendingTime - 1000) {
							// this rej shall consume inSeqNo -> in.onMessage
							throw new FixSessionException(SessionRejectReason.SENDINGTIME_ACCURACY_PROBLEM, "OrigSendingTime later than SendingTime".getBytes(), FixTags.ORIGSENDINGTIME_INT, FixUtils.getMsgType(msg.msgType));
						}
					}
				}
			}

			if (!Utils.equals(msg.targetCompID, url.getTargetCompID())) 
				throw new FixSessionException(SessionRejectReason.COMPID_PROBLEM, "Invalid targetCompID".getBytes(), FixTags.TARGETCOMPID_INT, FixUtils.getMsgType(msg.msgType));
			
			if (!Utils.equals(msg.senderCompID, url.getSenderCompID())) 
				throw new FixSessionException(SessionRejectReason.COMPID_PROBLEM, "Invalid senderCompID".getBytes(), FixTags.SENDERCOMPID_INT, FixUtils.getMsgType(msg.msgType));
			
			if (Boolean.getBoolean("fix.doSendingTimeCheck")) {
				long sendingTime = Utils.utcTimestampConverter.convert(msg.sendingTime).getTime();
				if ( sendingTime > plex.currentTimeMillis() + 120000 || sendingTime < plex.currentTimeMillis() - 120000)
					throw new FixSessionException(SessionRejectReason.SENDINGTIME_ACCURACY_PROBLEM, "Sendingtime accuracy problem".getBytes(), FixTags.SENDINGTIME_INT, FixUtils.getMsgType(msg.msgType));
			}
			// check other header stuff
			
			
			return true;
		}

		private void pendingInbound(ByteBuffer buf) {
			// we can only have one (the last) inbound message in the air, the rest w'll ask for in a resend
			ByteBuffer pending = ByteBuffer.allocate(buf.limit());
			pending.put((ByteBuffer)buf.flip());
			pending.flip();
			pendingInboundMsg.add(pending);
		}
	}

	private class TestRequestTimerTask implements ITimerTask {
		byte[] stamp = new byte[FixUtils.FIX_MAX_STRING_LENGTH]; //new byte[FixUtils.FIX_MAX_DIGITS];
		
		@Override
		public void run() {
			if (plex.currentTimeMillis() - lastInMsgTimeInMillis < url.getHeartBtInt() + url.getHeartBtInt() / 5 ) {
				testRequestTimerInstance.cancel(this);
				testRequestTimerInstance = plex.scheduleTimer(this, url.getHeartBtInt() - (plex.currentTimeMillis() - lastOutMsgTimeInMillis));
			} else if (outstandingHeartbeats == 0) {
				// first real timeout, send magic testrequest and restart the timer.
				Utils.longToNumeric(stamp, 0, plex.currentTimeMillis() + url.getHeartBtInt(), FixUtils.FIX_MAX_DIGITS / 2 + 2);
				stamp[stamp.length - 1] = FixUtils.SOH;
				in.onFixTestRequest(createTestRequest(stamp));
				outstandingHeartbeats++;
			} else {
				// second... ohps, magic logout that shall be replicated.
				byte[] text = ("Logout upon no response test request: currentTime=" + plex.currentTimeMillis() + " lastMessageTime=" + lastInMsgTimeInMillis + " ").getBytes();
				text[text.length - 1] = FixUtils.SOH;
				in.onFixLogout(createLogout(text));
				outstandingHeartbeats = 0;
			}
			
		}
		
	}
	
	private class HbtTimerTask implements ITimerTask {
		byte[] stamp = new byte[FixUtils.FIX_MAX_STRING_LENGTH]; //new byte[FixUtils.FIX_MAX_DIGITS];
		
		@Override
		public void run() {
			if (getFixSessionState() != FixSessionState.ACTIVE) return;
			
			if (plex.currentTimeMillis() - lastOutMsgTimeInMillis < url.getHeartBtInt() ) {
				hbtTimerInstance.cancel(this);
				hbtTimerInstance = plex.scheduleTimer(this, url.getHeartBtInt() - (plex.currentTimeMillis() - lastOutMsgTimeInMillis));
			} else {
				Utils.fill(stamp, (byte)0);
				stamp[stamp.length - 1] = FixUtils.SOH;
				in.onFixHeartbeat(createHeartbeat(stamp));
			}
			
		}
		
	}	
	
	private class DisconnectTimerTask implements ITimerTask {

		@Override
		public void run() {
			connection.close();
			setFixSessionState(FixSessionState.DISCONNECTED_CONNECTION_TODAY);
		}
		
	}
	
	abstract public class OutboundFixMessageListener extends FixMessageListenerImpl {
		// OUTBOUND MESSAGES - The FixSessionBase should be extended with a FIX version specific application message handler for the session layer.
		
		@Override
		public void onUnknownMessageType(FixMessage msg) {
			// ugh! Ignore
		}

		@Override
		public void onFixHeartbeat(FixHeartbeat msg) {
			send(msg);
			if (hbtTimerInstance!=null) hbtTimerInstance.cancel(hbtTimerTask);
			plex.scheduleTimer(hbtTimerTask, url.getHeartBtInt());
		}

		@Override
		abstract public void onFixLogon(FixLogon msg);
		
		@Override
		public void onFixLogout(FixLogout msg) {
			send(msg);
			if (getFixSessionState() != FixSessionState.RECEIVE_LOGOUT_MESSAGE) {
				setFixSessionState(FixSessionState.INITIATE_LOGOUT_PROCESS);
				if (logoutTimerInstance!=null) logoutTimerInstance.cancel(logoutTimerTask);
				logoutTimerInstance = plex.scheduleTimer(logoutTimerTask, url.getHeartBtInt());
			} else {
				connection.close();
				setFixSessionState(FixSessionState.DISCONNECTED_CONNECTION_TODAY);
			}
		}

		@Override
		public void onFixReject(FixReject msg) {
			send(msg);
		}

		@Override
		public void onFixResendRequest(FixResendRequest msg) {
			send(msg);
			setFixSessionState(FixSessionState.PROCESSING_RESPONSE_TO_RESENDREQUEST);
		}

		@Override
		public void onFixSequenceReset(FixSequenceReset msg) {
			send(msg);
		}

		@Override
		public void onFixTestRequest(FixTestRequest msg) {
			send(msg);
			if (testRequestTimerInstance!=null) testRequestTimerInstance.cancel(testRequestTimerTask);
			testRequestTimerInstance = plex.scheduleTimer(testRequestTimerTask, url.getHeartBtInt() + url.getHeartBtInt() / 5);
			setFixSessionState(FixSessionState.PROCESSING_RESPONSE_TO_TESTREQUEST);
		}
		
	}

	
	@Override
	public void onUnknownMessageType(FixMessage msg) {
		incrementInMsgSeqNum();
		out.onFixReject(createReject(SessionRejectReason.INVALID_MSGTYPE, "Invalid MsgType".getBytes(), FixTags.MSGTYPE_INT, FixUtils.getMsgType(msg.msgType)));
	}

	@Override
	public void onFixHeartbeat(FixHeartbeat msg) {
		if (isMagic(msg.testReqID)) {
			out.onFixHeartbeat(msg);
		} else {
			incrementInMsgSeqNum();
		}
	}

	@Override
	abstract public void onFixLogon(FixLogon msg);

	@Override
	public void onFixLogout(FixLogout msg) {
		
		if (isMagic(msg.text)) {
			out.onFixLogout(msg);
		} else {
			incrementInMsgSeqNum();
			if (getFixSessionState() == FixSessionState.INITIATE_LOGOUT_PROCESS) {
				logoutTimerInstance.cancel(logoutTimerTask);
				connection.close();
				setFixSessionState(FixSessionState.DISCONNECTED_CONNECTION_TODAY);
			} else {
				setFixSessionState(FixSessionState.RECEIVE_LOGOUT_MESSAGE);
			}
			out.onFixLogout(createLogout("Response to client logout".getBytes()));
		}
	}

	@Override
	public void onFixReject(FixReject msg) {
		incrementInMsgSeqNum();
		// magic text ends with SOH it is internal, sendit as a outbound
		if (isMagic(msg.text)) {
			out.onFixReject(msg);
			if (msg.sessionRejectReason == SessionRejectReason.COMPID_PROBLEM || msg.sessionRejectReason == SessionRejectReason.SENDINGTIME_ACCURACY_PROBLEM)
				out.onFixLogout(createLogout(msg.text));
		}
		
	}

	@Override
	public void onFixResendRequest(FixResendRequest msg) {
		incrementInMsgSeqNum();
		if ( msg.beginSeqNo >= outMsgSeqNum ) {
			out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.VALUE_IS_INCORRECT_OUT_OF_RANGE_FOR_THIS_TAG, 
					"BeginSeqNo must be less than current MsgSeqNum ".getBytes(), FixTags.BEGINSEQNO_INT, FixMessageInfo.MsgTypes.RESENDREQUEST) );
		} else if ( msg.endSeqNo != 0 && msg.endSeqNo > outMsgSeqNum ) {
			out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.VALUE_IS_INCORRECT_OUT_OF_RANGE_FOR_THIS_TAG, 
					"EndSeqNo must be 0 or greater than current MsgSeqNum ".getBytes(), FixTags.ENDSEQNO_INT, FixMessageInfo.MsgTypes.RESENDREQUEST) );
		} else {
			setFixSessionState(FixSessionState.HANDLE_RESENDREQUEST);
			FixMessage resend = null;
			int seqNo = (int) msg.beginSeqNo;
			long resendSeqNum = 0;
			resendBuf.clear();
			int read = 0;
			boolean doGapFill = false;
			while ((read = outboundLogHandle.read(resendBuf, seqNo, (int)msg.endSeqNo)) > 0) {
				try {
					while (resendBuf.hasRemaining()) {

						resend = resendParser.parse(resendBuf, resendListener);
						if (isReaplayableMessage(resend.msgType)) {
							if (doGapFill) {
								resend(createGapFill(resend.msgSeqNum - 1, resendSeqNum)); // the last admin msg is actually the previous message
								doGapFill = false;
							}
							resend(resend);
						} else {
							if (!doGapFill) resendSeqNum = resend.msgSeqNum;
							doGapFill = true;
						}
					}
				} catch (FixSessionException e) {
					out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.OTHER, 
							"Unable to process resend request ".getBytes(), e.refTagID, e.refMsgType) );
				} catch (FixGarbledException e) {
					out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.OTHER, 
							("Unable to process resend request " + e.getMessage()).getBytes(), Long.MAX_VALUE, new byte[0]) );
				} catch (IOException e) {
					out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.OTHER, 
							("IOException on write " + e.getMessage()).getBytes(), Long.MAX_VALUE, new byte[0]) );
				}
				seqNo+=read;
				resendBuf.clear();
			}
			
			if (doGapFill) {
				try {
					resend(createGapFill(resend.msgSeqNum, resendSeqNum));
				} catch (IOException e) {
					out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.OTHER, 
							("IOException on write " + e.getMessage()).getBytes(), Long.MAX_VALUE, new byte[0]) );
				}
				doGapFill = false;
			}
			setFixSessionState(FixSessionState.ACTIVE);
		}
	}

	@Override
	public void onFixSequenceReset(FixSequenceReset msg) {
		if (msg.newSeqNo > inMsgSeqNum + 1 && ( msg.msgSeqNum > inMsgSeqNum + 1 && msg.gapFillFlag )) {
			out.onFixResendRequest(createResendRequest(inMsgSeqNum + 1));
		} else if (msg.newSeqNo > inMsgSeqNum + 1 && ( msg.msgSeqNum == inMsgSeqNum + 1 && msg.gapFillFlag )) {
			inMsgSeqNum = (int) msg.newSeqNo - 1; // inMsgSeqNum is last received, eg. next -1
		} else if (msg.newSeqNo > inMsgSeqNum + 1 && msg.msgSeqNum < inMsgSeqNum + 1 && msg.possDupFlag && msg.gapFillFlag) { 
			// ignore
		} else if (msg.newSeqNo > inMsgSeqNum + 1 && msg.msgSeqNum < inMsgSeqNum + 1 && !msg.possDupFlag && msg.gapFillFlag) { 
			out.onFixLogout(createLogout(("MsgSeqNum too low, expecting " + (inMsgSeqNum + 1) + " received " + msg.msgSeqNum + " ").getBytes()));
			setFixSessionState(FixSessionState.INITIATE_LOGOUT_PROCESS);
		} else if (msg.newSeqNo < inMsgSeqNum + 1) {
			out.onFixReject(FixSessionBase.this.createReject(SessionRejectReason.VALUE_IS_INCORRECT_OUT_OF_RANGE_FOR_THIS_TAG, 
					"Value is incorrect (out of range) for this tag".getBytes(), FixTags.NEWSEQNO_INT, MsgTypes.SEQUENCERESET) );
		} else if (msg.newSeqNo == inMsgSeqNum + 1) { 
			// == strange reset, but accept it.
			inMsgSeqNum = (int) msg.newSeqNo - 1; // inMsgSeqNum is last received, eg. next -1
		} else if (msg.newSeqNo > inMsgSeqNum + 1) {
			// if == strange reset, but accept it.
			inMsgSeqNum = (int) msg.newSeqNo - 1; // inMsgSeqNum is last received, eg. next -1
		} 
	}

	@Override
	public void onFixTestRequest(FixTestRequest msg) {
		if (isMagic(msg.testReqID)) {
			out.onFixTestRequest(msg);
		} else {
			incrementInMsgSeqNum();
			out.onFixHeartbeat(FixSessionBase.this.createHeartbeat(msg.testReqID));
		}
	}
	
	
	
}

